package org.luxor.commons.zookeeper.config;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryNTimes;
import org.luxor.commons.zookeeper.client.ZookeeperClient;
import org.luxor.commons.zookeeper.client.ZookeeperJsonClient;
import org.luxor.commons.zookeeper.client.ZookeeperWatcherClient;
import org.luxor.commons.zookeeper.component.WatcherManager;
import org.luxor.commons.zookeeper.config.properties.ZookeeperProperties;
import org.luxor.commons.zookeeper.exception.ZookeeperException;
import org.luxor.commons.zookeeper.lock.GlobalLockTemplate;
import org.luxor.commons.zookeeper.lock.ZkLockAdvisor;
import org.luxor.commons.zookeeper.lock.ZkLockInterceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

/**
 * Zookeeper 全局配置
 *
 * @author Mr.Yan  @date: 2020/12/14
 */
@Configuration
@EnableConfigurationProperties(ZookeeperProperties.class)
@Import(ZookeeperListenerRegistrar.class)
@ConditionalOnProperty(prefix = ZookeeperProperties.PREFIX, name = "enabled", havingValue = "true", matchIfMissing = true)
public class ZookeeperAutoConfig implements DisposableBean {
    private static final Logger log = LoggerFactory.getLogger(ZookeeperAutoConfig.class);

    private ZookeeperProperties zookeeperProperties;
    private CuratorFramework curatorFramework;

    public ZookeeperAutoConfig(ZookeeperProperties zookeeperProperties) {
        Assert.notNull(zookeeperProperties, "Spring容器中找不到[zookeeperProperties]的实例化对象!");
        this.zookeeperProperties = zookeeperProperties;
    }

    @Bean
    @ConditionalOnMissingBean(CuratorFramework.class)
    public CuratorFramework curatorFramework() {
        log.info("Initialize Zookeeper Client Started...");
        try {
            log.debug("Initialize Zookeeper Client. host: {}", zookeeperProperties.getHost());

            // 重试策略
            int retryConnCount = zookeeperProperties.getRetryConnCount();
            int retryConnInterval = zookeeperProperties.getRetryConnInterval();
            RetryPolicy retryPolicy = new RetryNTimes(retryConnCount, retryConnInterval);

            String scheme = zookeeperProperties.getScheme();
            String auth = zookeeperProperties.getAuth();

            log.debug("Initialize Zookeeper Client. scheme:{} auth:{}", scheme, auth);

            // 创建客户端实例
            if (StringUtils.hasText(auth)) {
                this.curatorFramework = CuratorFrameworkFactory.builder().retryPolicy(retryPolicy)
                        .authorization(scheme, auth.getBytes())
                        .connectString(zookeeperProperties.getHost())
                        .sessionTimeoutMs(zookeeperProperties.getSessionTimeout())
                        .connectionTimeoutMs(zookeeperProperties.getConnectionTimeout())
                        .namespace(zookeeperProperties.getNameSpace())
                        .build();
            } else {
                this.curatorFramework = CuratorFrameworkFactory.builder().retryPolicy(retryPolicy)
                        .connectString(zookeeperProperties.getHost())
                        .sessionTimeoutMs(zookeeperProperties.getSessionTimeout())
                        .connectionTimeoutMs(zookeeperProperties.getConnectionTimeout())
                        .namespace(zookeeperProperties.getNameSpace())
                        .build();
            }
            // 打印异常日志
            curatorFramework.getUnhandledErrorListenable().addListener((message, e) -> {
                log.error(message, e);
            });

            // 启动客户端连接
            this.curatorFramework.start();

            log.info("Initialize Zookeeper Client completed.");
            return this.curatorFramework;
        } catch (Exception e) {
            log.error("Initialize Zookeeper Client fail.", e);
            throw new ZookeeperException("Initialize Zookeeper Client fail.", e);
        }
    }


    @Bean
    @ConditionalOnClass({CuratorFramework.class, ZookeeperProperties.class})
    @ConditionalOnMissingBean(WatcherManager.class)
    public WatcherManager watcherManager(CuratorFramework curatorFramework, ZookeeperProperties zookeeperProperties) {
        return new WatcherManager(curatorFramework, zookeeperProperties);
    }


    @Bean
    @ConditionalOnClass({CuratorFramework.class, WatcherManager.class})
    @ConditionalOnMissingBean(ZookeeperJsonClient.class)
    public ZookeeperJsonClient zookeeperJsonClient(CuratorFramework curatorFramework, WatcherManager watcherManager) {
        return new ZookeeperJsonClient(curatorFramework, watcherManager);
    }

    @Bean
    @ConditionalOnClass({CuratorFramework.class, WatcherManager.class})
    @ConditionalOnMissingBean(ZookeeperClient.class)
    public ZookeeperClient zookeeperClient(CuratorFramework curatorFramework, WatcherManager watcherManager) {
        return new ZookeeperClient(curatorFramework, watcherManager);
    }

    @Bean
    @ConditionalOnClass({CuratorFramework.class, WatcherManager.class})
    @ConditionalOnMissingBean(ZookeeperWatcherClient.class)
    public ZookeeperWatcherClient zookeeperWatcherClient(CuratorFramework curatorFramework
            , WatcherManager watcherManager) {
        return new ZookeeperWatcherClient(curatorFramework, watcherManager);
    }

    @Bean
    @ConditionalOnClass(CuratorFramework.class)
    @ConditionalOnMissingBean(GlobalLockTemplate.class)
    public GlobalLockTemplate globalLockTemplate(CuratorFramework curatorFramework) {
        return new GlobalLockTemplate(curatorFramework);
    }

    @Bean
    @ConditionalOnClass(CuratorFramework.class)
    public ZkLockInterceptor zkLockInterceptor(CuratorFramework curatorFramework) {
        return new ZkLockInterceptor(curatorFramework);
    }

    @Bean
    @ConditionalOnClass(ZkLockInterceptor.class)
    public ZkLockAdvisor zkLockAdvisor(ZkLockInterceptor zkLockInterceptor) {
        return new ZkLockAdvisor(zkLockInterceptor);
    }

    @Override
    public void destroy() throws Exception {
        log.info("Close the zookeeper connect  Started...");
        curatorFramework.close();
        log.info("Close the zookeeper connect completed.");
    }
}
